package com.lzt.test

import java.util.{Date, Properties}

import com.lzt.app.StringUtil
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

object TestKafkaProducet {

  def main(args: Array[String]): Unit = {


    var topic = "spark"
    val props = new Properties()
    props.put("bootstrap.servers", "localhost:9092")
    props.put("group.id", "consumer-tutorial")
    props.put("key.serializer", classOf[StringSerializer].getName)
    props.put("value.serializer", classOf[StringSerializer].getName)
    //    props.put("acks", "all")
    val producer = new KafkaProducer[String, String](props)
    while (true) {
      val record = new ProducerRecord[String, String](topic, s"tpm-${System.currentTimeMillis()}", "{\"topic\":\"tccgl\",\"type\":\"delay\",\"time\":\"2018-07-31 12:12:12\",\"data\":{\"avg\":\"15.769\",\"fail\":\"0\",\"id\":\"7\",\"ip\":\"192.168.1.8\",\"max\":\"25.004\",\"min\":\"13.318\",\"name\":\"翔飞停车场\",\"online\":\"true\",\"ping\":\"10\",\"suc\":\"10\",\"time\":\"201807111801\",\"rowkey\":\"201807111801-7\"}}")
      //异步写
      val future = producer.send(record)
      //    import org.apache.kafka.clients.producer.RecordMetadata
      //    val metadata = future.get
      //    println(metadata.toString)
      //同步写
      producer.flush()
      Thread.sleep(3000)
    }
  }
}
